package com.newland.nxsms;

import com.newland.nxsms.cmpp.CMPPDao;
import com.newland.nxsms.cmpp.CMPPTaskInfo;
import com.newland.nxsms.dao.ConnectionManager;
import com.newland.nxsms.utils.RedisCluster;
import com.newland.nxsms.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class AlreadySendDataToDb implements Runnable {
	private static final Logger logger = LoggerFactory.getLogger(AlreadySendDataToDb.class);
	private CMPPDao dao;
	private RedisCluster cluster;
	public AlreadySendDataToDb(CMPPDao dao) {
		this.dao = dao;
	}
	private AtomicLong atomicLong = new AtomicLong(0L);

	public AtomicLong getAtomicLong() {
		return atomicLong;
	}
	public void setAtomicLong(Long i) {
		this.atomicLong.set(i); ;
	}
	@Override
	public void run() {
		try {
			cluster = new RedisCluster(NxSmsMain.REDIS_CFG);
		} catch (IOException e) {
			logger.error("send data to db redis init failed:error "+ e.getMessage());
			e.printStackTrace();
		}
		int insertCount = 0;
		List<CMPPTaskInfo> insertList = new ArrayList<CMPPTaskInfo>();
		while (true) {
			try {
				CMPPTaskInfo cmppTaskInfo = null;
				cmppTaskInfo = NxSmsMain.insertSqls.poll();
				if (null != cmppTaskInfo) {
					this.getAtomicLong().incrementAndGet();
					insertCount++;
					Integer seqId = cmppTaskInfo.getSeqId();
					String state = cluster.get(NxSmsMain.HOST_NAME+ "updateReport" + seqId);
					if (StringUtils.isNotBlank(state)){
						cmppTaskInfo.setState(Integer.parseInt(state));
						cluster.expire(NxSmsMain.HOST_NAME+"updateReport" + seqId,0);
					} else {
						cmppTaskInfo.setState(0);
					}

					insertList.add(cmppTaskInfo);
					if (insertCount == NxSmsMain.insertUp) {
						Connection conn = ConnectionManager.getInstance().getConnection();
						Long aLong = dao.updateInsertMsgState(conn, true, insertList);
						if (aLong>0){
							logger.info("insertSmsReport one time cost :"+aLong);
						}
						insertList.clear();
						insertCount = 0;
					}
				} else {
					Thread.sleep(100);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
